/*************************************************************************
	> File Name: acquisitor.cpp
	> Author: xuwenlong
	> Mail: myxuan475@126.com 
	> Created Time: 2018年02月02日 星期五 14时22分15秒
 ************************************************************************/
#include<stdio.h>
#include <dlfcn.h>
#include <time.h>
#include <utils/sqlite.h>
#include <dbms/acquire.h>
#include <dbms/exprs.h>
#include <algorithm>
#include <dbms/status.h>
#include "acqdata.h"
#include "collector.h"
#include "acquisition.h"
#include <devices/devices.h>
#include <dbms/energy.h>

Collector::Collector(const deviceinfo_t& devinfo):
    Acquisitor(devinfo),
    m_iDataLock(TRUE),
    m_fixlistData(devinfo.id),
    m_last(0),
    m_uCommitId(0) {

    for (int i = 0;i < MAX_TASK_DIVID;i++) {
        m_uCommits[i] = -1;
    }
}

Collector::~Collector()
{
}

void Collector::Run(
        tConnection connector,
        const time_t &acqtime,
        vector<AcqItem> &listAcquire)
{
    int ret = 0;
    if (listAcquire.empty() ||
            acqtime-m_last > 180) {
        m_last = acqtime;//更新时间
        listAcquire.clear();
        if (0 != (ret = acquireData(connector,listAcquire,24))) {
            m_bStat = FALSE;
            return commitError(GetError((DAS_RESULT)ret));
        }
    }

    AutoLock autolock(m_iDataLock);
    for (vector<AcqItem>::iterator iter = listAcquire.begin();
         iter != listAcquire.end();iter++) {
        AcqItem &acqItem = (AcqItem &)*iter;
        if (acqItem.GetId()/10000 != 3 &&
                acqItem.GetId()/10000 != 4)
            continue;
        AcqData acqData = acqItem;
        acqData.SetTime(acqtime);
        m_fixlistData.push_back(acqData);
    }
    m_bStat = TRUE;
}

U16 Collector::acquireData(
        tConnection connector,
        vector<AcqItem> &listAcquire,
        U8 flag)
{
    tDataItem acqItem;
    tDataItem bfItem;
    vector<CmdValue> listResult;

    if (!connector.valid())
        return DAS_FAILDCONNECTION;

    bfItem = NULL;
    acqItem = GetAcqItem(NULL,flag);
    if (!acqItem.valid())
        return DAS_NOITEM;

    do {
        OpValue result;
        U16 err;
        if (!acqItem->isCmdEqual(bfItem) || listResult.empty()) {
            //重新采集
            listResult.clear();
            err = acquireDevice(connector,acqItem,result,&listResult);
        }
        else {
            //使用之前的
            S32 funcId = 0;
            err = 0;

            acqItem->ResetCmd();
            for (vector<CmdValue>::iterator iter = listResult.begin();
                 iter != listResult.end();iter++) {
                CmdValue recv = *iter;
                RECV_RESULT r;
                r = acqItem->ParseRecvCmd(
                            recv,funcId++);

                if (RECV_COMPLETE != r) {
                    err = DAS_PARSEFAILED;
                    break;
                }

            }
            if (err != 0) {
                return err;
            }
            result = acqItem->Result();
        }

        if (err != 0) {
            return err;
        }
        AcqItem acqdata = result;
        acqdata.SetId(acqItem->GetItemId());
        acqdata.SetName(acqItem->GetName());

        listAcquire.push_back(acqdata);

        bfItem = acqItem;
        acqItem = GetAcqItem(acqItem,flag);
    }while(acqItem.valid());

    //备份数据
    return 0;
}

void Collector::handleJsonResult(BOOL ret,S32 si)
{
    LOG_TRACE();
    AutoLock autolock(m_iDataLock);

    m_uCommitId &= ~(0x01<<si);
    if (si<0 || si >= MAX_TASK_DIVID)
        return;

    int size = m_fixlistData.size();
    int index = m_uCommits[si]+1;
    if (index >= size)
        return;

    while (index < size) {
        AcqData tmp = m_fixlistData.at(index);
        if (!tmp.IsCommit(si))
            break;
        if (ret) {
            tmp.SetVerify(si);
            m_uCommits[si] = index;
            if (tmp.IsAllVerify(TotalCenter())) {
                m_fixlistData.remove(index);
                size--;
                for (int i = 0;i < TotalCenter();i++) {
                    if (m_uCommits[i] >= 0)
                        m_uCommits[i]--;
                }
            }
            else {
                m_fixlistData.replace(index,tmp);
                index++;
            }
        }
        else {
            tmp.ResetFlag(si);
            m_fixlistData.replace(index,tmp);
            index++;
        }
    }

    LOG_DEBUG("serId = %d,ret = %d,id = %d,from size = %d to %d",
              si,ret,m_idevinfo.id,size,m_fixlistData.size());
}

BOOL Collector::OnReceiveJson(S32 index,
        DAS_RESULT result,
        const string &method,
        const Json::Value &json)
{
    if (method == "acquireData") {
        if (result == DAS_SUCCESS && json.asBool() == true) {
            handleJsonResult(true,index);
        }
        else {
            handleJsonResult(false,index);
        }
    }
    return TRUE;
}

S32 Collector::commitData(S32 index)
{
    AutoLock aulock(m_iDataLock);

    if (m_uCommitId & (0x01<<index))
        return -1;
    Json::Value params;
    if (!fetchCommitData(index,params))
        return -1;

    int commitId = -1;
    {
        AutoUnlock unlock(m_iDataLock);
        commitId = sendRpc(index,"acquireData",params);
    }while(0);

    if (commitId < 0) {
        resetCommit(index);
        LOG_ERR("serId = %d,commit failed:devId = %04d",
                index,
                m_idevinfo.id);
        return -1;
    }
    m_uCommitId |= 0x01<<index;
    LOG_DEBUG("serId = %d,new commit:devId = %04d,msgId = %d",
              index,
              m_idevinfo.id,commitId);
    return 0;
}

void Collector::clearData()
{
    AutoLock autolock(m_iDataLock);
    m_fixlistData.clear();
}

bool Collector::hasItemId(S32 id)
{
    tDataItem item = GetItemById(id);
    if (item.valid())
        return true;
    if (id/10000 == 0) {
        id = 30000+id;
        item = GetItemById(id);
        return item.valid();
    }
    return false;
}

void Collector::resetCommit(S32 si)
{
    int index = m_uCommits[si]+1;
    int size = m_fixlistData.size();

    while (index < size ) {
        AcqData tmp = m_fixlistData.at(index);
        if (tmp.IsVerify(si)) {
            m_uCommits[si] = index;
            continue;
        }
        if (tmp.IsCommit(si)) {
            tmp.ResetFlag(si);
            m_fixlistData.replace(index,tmp);
        }
        else
            break;
        index++;
    }
}

bool Collector::fetchCommitData(int si,Json::Value &params)
{
    AutoLock autolock(m_iDataLock);
    if (m_fixlistData.isEmpty()) {
//        LOG_DEBUG("%04d(%d) is empty",m_idevinfo.id,m_iCommitId);
        return false;
    }

    resetCommit(si);

    int size = m_fixlistData.size();
    int total = 0;
    for (int i = m_uCommits[si]+1;i < size;i++) {
        AcqData tmp = m_fixlistData.at(i);

        if (!hasItemId(tmp.itemid)) {
            m_fixlistData.remove(i);
            LOG_DEBUG("data bad:id:%d,time=%s,flag=%d,data=%s",
                      tmp.itemid,utils::timetostr(tmp.GetTime()).c_str(),
                      tmp.flag,tmp.ToJson().toStyledString().c_str());
            i--;
            size--;
            continue;
        }

        Json::Value jsondata;
        Json::Value records;
        records["id"] = m_idevinfo.id;
        records["time"] = utils::timetostr(tmp.GetTime());

        string stracq = acquire::idToStr(tmp.itemid);

        jsondata[stracq] = tmp.ToJson();
        tmp.SetCommit(si);
        m_fixlistData.replace(i,tmp);
        total++;

        int k = i;
        while (++k < size) {
            AcqData item = m_fixlistData.at(k);

            if (item.GetTime() != tmp.GetTime()) {
                --k;
                break;
            }
            if (!hasItemId(item.itemid)) {
                m_fixlistData.remove(k);
                size--;
                LOG_DEBUG("data bad:id:%d,time=%s,flag=%d,data=%s",
                          item.itemid,utils::timetostr(item.GetTime()).c_str(),
                          item.flag,item.ToJson().toStyledString().c_str());
                continue;
            }

            stracq = acquire::idToStr(item.itemid);
            jsondata[stracq] = item.ToJson();
            item.SetCommit(si);
            m_fixlistData.replace(k,item);
            total++;
        }
        records["data"] = jsondata;
        params.append(records);
        i = k;
        if (total>20)
            break;
    }
    return true;
}
